// --------------------------------------------------------------------------------------------------------------------
// <copyright file="FilePlayerAgent.cs" company="Microsoft Corporation">
// The MIT License (MIT)
// 
// Copyright (c) 2014, Microsoft Corporation
// 
// Permission is hereby granted, free of charge, to any person obtaining a copy
//  of this software and associated documentation files (the "Software"), to deal
//  in the Software without restriction, including without limitation the rights
//  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//  copies of the Software, and to permit persons to whom the Software is
//  furnished to do so, subject to the following conditions:
// 
// The above copyright notice and this permission notice shall be included in
//  all copies or substantial portions of the Software.
// 
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//  THE SOFTWARE.
// </copyright>
// --------------------------------------------------------------------------------------------------------------------
namespace Microsoft.Robotics.Runtime
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Linq;
    using System.Runtime.Serialization;
    using System.Threading;

    /// <summary>
    /// Agent that plays back a file of any kind.
    /// </summary>
    [DataContract]
    [System.Diagnostics.CodeAnalysis.SuppressMessage(
        "Microsoft.Design",
        "CA1001:TypesThatOwnDisposableFieldsShouldBeDisposable",
        Justification = "PipelineElements clean up resources when OnDeactivated() is called")]
    public abstract class FilePlayerAgent : ConsumerProducerAgent<TimerMessage, CompletionMessage>, IMessagePublisher 
    {
        /// <summary>
        /// The maximum number of objects we read in advance, to make sure we have something to publish at all times.
        /// </summary>
        private const int BufferSize = 10;

        /// <summary>
        /// The path to the file to play back.
        /// </summary>
        [DataMember(Name = "Path")]
        private readonly string filePath;

        /// <summary>
        /// The types of messages published by the player.
        /// </summary>
        [DataMember(Name = "MessageTypes")]
        private readonly Type[] messageTypes;

        /// <summary>
        /// Enables initialization of recording start as time of first message in recording
        /// By default recording start time is set to 0. 
        /// Skip ahead to first message must not be enabled for multiple players
        /// </summary>
        [DataMember(Name = "SkipAheadToFirstMessage")]
        private bool skipAheadToFirstMessage;

        /// <summary>
        /// Tracks if start time should skip to next message
        /// </summary>
        private bool doSkipAheadOnNextMessage;

        /// <summary>
        /// The IFileReader that can deserialize the messages
        /// </summary>
        private IFileReader reader;

        /// <summary>
        /// Collection of publishers
        /// </summary>
        private Dictionary<Type, IForwarder> publishers;

        /// <summary>
        /// The thread that does the reading of files.
        /// </summary>
        private Thread fileReaderThread;

        /// <summary>
        /// Indicates whether we finished reading (either due to reaching the end of file, or because we've been told to stop). 
        /// </summary>
        private ManualResetEvent completed;

        /// <summary>
        /// The buffer of deserialized messages waiting to be published.
        /// This publishing queue serves two purposes: 
        /// 1. allows us to read ahead so that if there are messages that are due at the same time, we can still publish them at the right time because we deserialized them in advance
        /// 2. limits the number of messages we read ahead to the value given by BufferSize, to avoid gobbling too much memory (some messages, e.g. video frames, are particularly large)
        /// </summary>
        private BlockingCollection<AgentMessage> publishingQueue;

        /// <summary>
        /// The next message to publish.
        /// </summary>
        private AgentMessage nextMessage;

        /// <summary>
        /// The time origin for the recorded messages. 
        /// </summary>
        private long recordingStartTime;

        /// <summary>
        /// Indicates whether we are still reading the file or not
        /// </summary>
        private bool isReading;

        /// <summary>
        /// Initializes a new instance of the FilePlayerAgent class
        /// </summary>
        /// <param name="name">The name of the player agent.</param>
        /// <param name="filePath">The path to the file to play back.</param>
        /// <param name="messageTypes">The types of messages played back by this agent.</param>
        /// <param name="timerAgent">The timer agent to subscribe to. This needs to be a 1ms timer.</param>
        /// <param name="skipAheadToFirstMessage">Flag indicating if recording start time is initialized to first message time. 
        /// This should not be enabled for multiple players</param>
        public FilePlayerAgent(string name, string filePath, Type[] messageTypes, IProducer<TimerMessage> timerAgent, bool skipAheadToFirstMessage = false)
            : base(name, timerAgent.Name)
        {
            this.filePath = filePath;
            this.messageTypes = messageTypes;
            this.skipAheadToFirstMessage = skipAheadToFirstMessage;
            
            // Wait for activation to enable skip ahead when required
            this.doSkipAheadOnNextMessage = false;
        }

        /// <summary>
        /// Gets the set of types we intend to publish
        /// </summary>
        public IEnumerable<Type> PublishedTypes
        {
            get { return this.messageTypes; }
        }

        /// <summary>
        /// Called to do our own activation before the control loop is (re)started
        /// </summary>
        public override void OnActivated()
        {
            // we do all initialization in here because we expect the same agent instance to be deactivate and reactivated many times with the intent of restating playback
            this.completed = new ManualResetEvent(false);
            this.publishingQueue = new BlockingCollection<AgentMessage>(new ConcurrentQueue<AgentMessage>(), BufferSize);
            this.nextMessage = null;
            this.isReading = true;

            this.reader = this.CreateReader();
            this.recordingStartTime = this.reader.Initialize(this.filePath, this.publishingQueue);
            
            // Start a background thread to deserialize messages in advance
            this.fileReaderThread = new Thread(this.reader.Start);
            this.fileReaderThread.Name = this.Name;
            this.fileReaderThread.Start();

            // Enable skipping on first message if required
            this.doSkipAheadOnNextMessage = this.skipAheadToFirstMessage;

            base.OnActivated();
        }

        /// <summary>
        /// Called to do our own cleanup right after the control loop is stopped
        /// </summary>
        public override void OnDeactivated()
        {
            this.reader.Stop();
            this.fileReaderThread.Join();
            this.publishingQueue = null;
            this.completed.Dispose();
            base.OnDeactivated();
        }

        /// <summary>
        /// Called to pass in all publishing ports for the supported PublishedTypes
        /// </summary>
        /// <param name="forwarders">The set of publishing ports to use</param>
        public void InitializePublishers(Dictionary<Type, IForwarder> forwarders)
        {
            this.publishers = forwarders;
        }

        /// <summary>
        /// Used by the manifest generator. Makes a promise about being a publisher of T.
        /// A regular producer agent is an IProducer through implementation. 
        /// The FilePlayerAgent is special in that it doesn't have this information statically.
        /// </summary>
        /// <typeparam name="T">The type of message</typeparam>
        /// <returns>An IProducer for the given message type</returns>
        public IProducer<T> AsProducer<T>()
            where T : AgentMessage
        {
            if (!this.messageTypes.Contains(typeof(T)) && typeof(T) != typeof(CompletionMessage))
            {
                throw new InvalidCastException();
            }

            ProxyDefinition.ProxyProducer<T> producer = new ProxyDefinition.ProxyProducer<T>();
            producer.Name = this.Name;
            return producer;
        }

        /// <summary>
        /// Called when a new timer tick arrives.
        /// </summary>
        /// <param name="message">The timer message</param>
        public override void Receive(TimerMessage message)
        {
            bool isFileReaderThreadAlive = this.fileReaderThread.IsAlive;

            // look for queued messages ready to be published.
            bool hasMessages = this.isReading;
            while (hasMessages)
            {
                hasMessages = false;
                if (this.nextMessage == null)
                {
                    // try to read another message, blocking until we get one (or until the collection is marked as complete)
                    // to make sure we are not running behind
                    // Without this blocking wait, multiple file readers could get out of sync when fast-forwarding
                    // because beyond a certain speed they all run behind (deserialization becomes the bottleneck)
                    // The slowest one (i.e. the file with the largest messages) would run further and further behind
                    // Blocking here will cause the timer to block (in offline mode, which should always be used when playing back files) 
                    // preventing the players with smaller message sizes from advancing too fast
                    this.publishingQueue.TryTake(out this.nextMessage, -1);
                }

                if (this.nextMessage != null)
                {
                    if (true == this.doSkipAheadOnNextMessage)
                    {
                        this.recordingStartTime = this.nextMessage.Timestamp;
                        this.doSkipAheadOnNextMessage = false;
                    }

                    // if we have a message and it's due, publish it
                    long relativeTimestamp = this.nextMessage.Timestamp - this.recordingStartTime;
                    if (relativeTimestamp <= message.RelativeTime)
                    {
                        this.publishers[this.nextMessage.GetType()].Post(this.nextMessage);
                        this.nextMessage = null;
                        hasMessages = true;
                    }
                }
            }

            // if the file reading thread was already stopped when we started and we published all the messages in the queue, we are done
            if (this.isReading && !isFileReaderThreadAlive && this.nextMessage == null)
            {
                this.completed.Set();
                CompletionMessage msg = new CompletionMessage(this.Name);
                this.Publisher.Post(msg);
                this.isReading = false;
            }
        }

        /// <summary>
        /// Waits for the playback to complete. Only used by tests.
        /// </summary>
        /// <param name="timeout">The amount of time to wait, in ms, before giving up.</param>
        /// <returns>True if the wait was signaled, false if it timed out</returns>
        internal bool Wait(int timeout)
        {
            return this.completed.WaitOne(timeout);
        }

        /// <summary>
        /// Creates the right type of file reader. 
        /// </summary>
        /// <returns>A well-configured reader</returns>
        protected abstract IFileReader CreateReader();
    }
}
